package cn.sciento.transfer.app.service;

import cn.sciento.core.domain.Page;
import cn.sciento.core.exception.CommonException;
import io.choerodon.mybatis.pagehelper.PageHelper;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.ibatis.session.TransactionIsolationLevel;
import org.apache.rocketmq.tools.command.SubCommandException;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.mybatis.DynamicSqlMapper;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.ConsTenantConfig;
import cn.sciento.transfer.domain.entity.ProducerConfig;
import cn.sciento.transfer.domain.repository.ConsDbConfigRepository;
import cn.sciento.transfer.domain.repository.ConsTenantConfigRepository;
import cn.sciento.transfer.domain.repository.EventSendRepository;
import cn.sciento.transfer.domain.repository.ProducerConfigRepository;
import cn.sciento.transfer.infra.datasource.mybatis.DynamicSqlMapperTemplate;
import cn.sciento.transfer.infra.datasource.sql.SqlBuilder;
import cn.sciento.transfer.infra.datasource.sql.SqlUtils;
import cn.sciento.transfer.infra.datasource.utils.AsyncExecutionTemplate;
import cn.sciento.transfer.infra.datasource.utils.TransferDataSourceTemplate;
import cn.sciento.transfer.infra.feign.HpfmRemoteService;
import cn.sciento.transfer.infra.rocketmq.config.MqConfigProperties;
import cn.sciento.transfer.infra.rocketmq.service.MqConsumerService;
import cn.sciento.transfer.infra.rocketmq.service.impl.MqConsumerServiceImpl;
import cn.sciento.transfer.infra.rocketmq.utils.RocketmqClientUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

public abstract class AbstractDataInitService {
  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDataInitService.class);
  
  public static final String TABLE_NAME_TL_SUFFIX = "_tl";
  
  public static final String TABLE_NAME_BAK_SUFFIX = "_bak";
  
  public static final String TABLE_NAME_RESTORE_SUFFIX = "_restore";
  
  public static final String MASTER_TABLE_KEY = "master";
  
  @Autowired
  protected TransferDataSourceTemplate transferDataSourceTemplate;
  
  @Autowired
  protected ProducerConfigRepository producerConfigRepository;
  
  @Autowired
  protected ConsDbConfigRepository consDbConfigRepository;
  
  @Autowired
  protected ConsTenantConfigRepository consTenantConfigRepository;
  
  @Autowired
  protected EventSendRepository eventSendRepository;
  
  @Autowired
  private HpfmRemoteService hpfmRemoteService;
  
  @Autowired
  protected MqConsumerService mqConsumerService;
  
  @Autowired
  protected MqConfigProperties mqConfigProperties;
  
  @Autowired
  protected AsyncTaskExecutor asyncTaskExecutor;
  
  @Autowired
  private DynamicSqlMapperTemplate dynamicSqlMapperTemplate;
  
  public final void doInit(DataInitContext dataInitContext) {
    LOGGER.info("enter do init with data init context: {}.", dataInitContext);
    if (!validate(dataInitContext)) {
      return;
    }
    try {
      preInit(dataInitContext);
      if (BaseConstants.Flag.YES.equals(dataInitContext.getProducerConfig().getInitImportDataFlag())) {
        doInitialize(dataInitContext);
      }
      postInit(dataInitContext);
    } catch (Exception ex) {
      LOGGER.info("exit do init failed with final data init context: {}.", dataInitContext);
      LOGGER.error("exit do init with exception: ", ex);
      handleException(dataInitContext, ex);
    } finally {}
    LOGGER.info("exit do init success with final data init context: {}.", dataInitContext);
  }
  
  protected boolean validate(DataInitContext dataInitContext) {
    Assert.notNull(dataInitContext, "data init context cannot be null.");
    Assert.notNull(dataInitContext.getProducerConfig(), "producer config cannot be null.");
    Assert.notNull(dataInitContext.getConsDbConfig(), "consumer db config cannot be null.");
    ProducerConfig producerConfig = dataInitContext.getProducerConfig();
    ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
    try {
      consDbConfig.validInitHistoryDataStatus(producerConfig, this.consDbConfigRepository, true);
    } catch (Exception e) {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("valid sts error ", e);
      }
      return false;
    } 
    return true;
  }
  
  protected final void preInit(DataInitContext dataInitContext) {
    ProducerConfig producerConfig = dataInitContext.getProducerConfig();
    producerConfig.analyzeDDL();
    LOGGER.info("after analyze ddl, producer config is: {}.", producerConfig);
    preInitRunningStatus(dataInitContext);
    preInitTable(dataInitContext);
    preInitSuspendConsumerGroup(dataInitContext);
  }
  
  protected final void doInitialize(DataInitContext dataInitContext) {
    List<DataTransferContext> dataTransferContextList = doInitializeBuildDataTransferContext(dataInitContext);
    if (CollectionUtils.isEmpty(dataTransferContextList)) {
      return;
    }
    LOGGER.info("do initialize with data transfer context: {}.", dataTransferContextList);
    dataTransferContextList.forEach(this::internalDoDataTransfer);
  }
  
  protected final DataTransferContext internalBuildDataTransferContext(DataInitContext dataInitContext, TableInitContext sourceTableInitContext, TableInitContext targetTableInitContext, List<String> sourceTableInitFetchDataSqlWhereList, String sourceTableAppendingCustomSqlRawSlice, Object sourceTableInitFetchDataSqlParameterValue) {
    if (sourceTableInitContext == null) {
      return null;
    }
    sourceTableInitContext
      .buildQualifiedTableName()
      .buildInitFetchDataSql(sourceTableInitFetchDataSqlWhereList, sourceTableAppendingCustomSqlRawSlice, sourceTableInitFetchDataSqlParameterValue);
    targetTableInitContext.buildQualifiedTableName();
    return new DataTransferContext(dataInitContext, sourceTableInitContext, targetTableInitContext);
  }
  
  protected final void internalDoDataTransfer(DataTransferContext dataTransferContext) {
    Assert.notNull(dataTransferContext, "data transfer context cannot be null.");
    dataTransferContext.validate();
    AbstractDataInitService.TableInitContext sourceTableInitContext = dataTransferContext.getSourceTableInitContext();
    this.dynamicSqlMapperTemplate.doExecute(sourceTableInitContext.getInitDataSourceCode(), sourceTableInitContext.getInitServiceName(), TransactionIsolationLevel.REPEATABLE_READ, (dynamicSqlMapper) -> {
      ArrayList futureList = new ArrayList(16);
      dataTransferContext.dataInitContext.setDataInitTimestamp(new Date());
      Page<Map<String, Object>> firstPage = this.internalFetchPageData(dynamicSqlMapper, sourceTableInitContext, 0);
      int totalPages = firstPage.getTotalPages();
      futureList.add(this.internalDoDataTransferAsync(dataTransferContext, dataTransferContext.getTargetTableInitContext(), firstPage));

      for(int i = 1; i < totalPages; ++i) {
        Page<Map<String, Object>> currentPage = this.internalFetchPageData(dynamicSqlMapper, sourceTableInitContext, i);
        futureList.add(this.internalDoDataTransferAsync(dataTransferContext, dataTransferContext.getTargetTableInitContext(), currentPage));
      }

      try {
        List<Integer> affectedRowsList = new ArrayList(16);

        for (Object o : futureList) {
          Future future = (Future) o;
          affectedRowsList.addAll((Collection<? extends Integer>) future.get());
        }

        LOGGER.info("exit do data transfer with success, with data transfer context: {}, and affected rows: {}.", dataTransferContext, affectedRowsList.stream().reduce((prev, next) -> {
          return prev + next;
        }).orElse(-1));
        return null;
      } catch (Exception var10) {
        LOGGER.error("await async do data transfer result with error: ", var10);
        dataTransferContext.dataInitContext.stop();
        throw new CommonException(var10);
      }
    });
  }
  
  private Page<Map<String, Object>> internalFetchPageData(DynamicSqlMapper dynamicSqlMapper, TableInitContext tableInitContext, int page) {
    Assert.notNull(tableInitContext, "table init context cannot be null.");
    Assert.notNull(tableInitContext.getInitFetchDataSql(), "init fetch data sql cannot be null.");
    Assert.notNull(tableInitContext.getInitFetchDataSqlParameterValue(), "init fetch data sql parameter value cannot be null.");
    PageHelper.startPage(page, 20);
    List<Map<String, Object>> rawData = dynamicSqlMapper.selectList("<script>\n" + tableInitContext.getInitFetchDataSql() + "\n</script>", tableInitContext
        .getInitFetchDataSqlParameterValue());
    return (rawData == null) ? new Page() : (Page<Map<String, Object>>)rawData;
  }

  private Future<List<Integer>> internalDoDataTransferAsync(AbstractDataInitService.DataTransferContext dataTransferContext, AbstractDataInitService.TableInitContext targetTableInitContext, Page<Map<String, Object>> rawData) {
    return AsyncExecutionTemplate.asyncDoExecute(AsyncExecutionTemplate.findAsyncTaskExecutor("transferTargetTaskExecutor"), () -> {
      List<Integer> resultList = new ArrayList(32);
      if (CollectionUtils.isEmpty(rawData)) {
        LOGGER.info("raw data is empty, exit.");
        return resultList;
      } else if (!dataTransferContext.dataInitContext.canInit()) {
        LOGGER.info("data transfer has stoped. ignore current page data (page = {}) and after.", rawData.getNumber());
        return resultList;
      } else {
        int[] resultArray = this.transferDataSourceTemplate.doExecute(targetTableInitContext.getInitDataSourceCode(), targetTableInitContext.getInitServiceName(), (jdbcTemplate) -> {
          List<String> insertFieldList = new ArrayList(((Map)rawData.get(0)).keySet());
          String insertSql = SqlBuilder.buildInsert(insertFieldList, targetTableInitContext.getInitQualifiedTableName());
          return SqlUtils.doBatchUpdate(jdbcTemplate, insertSql, rawData);
        });
        LOGGER.info("============同步数据到消费表,服务名：【{}】，表名：【{}】，同步数据量：【{}】===========", targetTableInitContext.getInitServiceName(), targetTableInitContext.getInitQualifiedTableName(), resultArray.length);
        resultList.addAll(Arrays.stream(resultArray).boxed().collect(Collectors.toList()));
        return resultList;
      }
    });
  }
  
  protected final void postInit(DataInitContext dataInitContext) {
    postInitRefreshTargetTable(dataInitContext);
    postInitRefreshStatus(dataInitContext);
    postInitResumeConsumerGroup(dataInitContext);
  }
  
  private void postInitRefreshTargetTable(DataInitContext dataInitContext) {
    ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
    this.transferDataSourceTemplate.doExecuteInConnection(consDbConfig
        .getConsumerDs(), consDbConfig
        .getConsumerService(), consDbConfig
        .getConsumerDb(), connection -> {
          try {
            postInitRefreshTargetTableCustomize(connection, dataInitContext);
          } catch (Exception ex) {
            LOGGER.error("interchange production table and temp table with error: ", ex);
            throw new CommonException(ex);
          } 
          return null;
        });
  }
  
  private void postInitResumeConsumerGroup(DataInitContext dataInitContext) {
    ProducerConfig producerConfig = dataInitContext.getProducerConfig();
    ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
    ConsTenantConfig consTenantConfig = dataInitContext.getConsTenantConfig();
    if (!this.mqConsumerService.hasBeenStarted(consDbConfig, consTenantConfig)) {
      this.mqConsumerService.startListener(producerConfig
          .getServiceCode(), producerConfig
          .getTableName(), consDbConfig, consTenantConfig);
    } else {
      this.mqConsumerService.resume(consDbConfig, consTenantConfig);
    } 
  }
  
  protected final Long internalBuildOffset(DataInitContext dataInitContext, Long tenantId) {
    ProducerConfig producerConfig = dataInitContext.getProducerConfig();
    Long eventId = this.eventSendRepository.selectEventByTimeStamp(dataInitContext
        .getDataInitTimestamp(), producerConfig
        .getServiceCode(), producerConfig
        .getTableName(), tenantId);
    if (eventId == null) {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("===========NO MSG OFFSET FOUND:[time={}]===============", dataInitContext.getDataInitTimestamp());
      }
      return null;
    } 
    try {
      Long offset = RocketmqClientUtils.queryMessageOffsetByKey(this.mqConfigProperties
          .getNameSrv(), 
          MqConsumerServiceImpl.createTopicName(producerConfig.getServiceCode(), producerConfig.getTableName()), 
          String.valueOf(eventId));
      return offset;
    } catch (SubCommandException ex) {
      LOGGER.info(">>>>>>>>>no message found, key[{}], nameSrc[{}]<<<<<<<<<<<<<", eventId, this.mqConfigProperties.getNameSrv());
      LOGGER.error("fetch mq current offset failed: ", (Throwable)ex);
      return null;
    } 
  }
  
  protected final void handleException(DataInitContext dataInitContext, Exception ex) {
    ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
    ConsTenantConfig consTenantConfig = dataInitContext.getConsTenantConfig();
    this.mqConsumerService.resume(consDbConfig, consTenantConfig);
    try {
      this.transferDataSourceTemplate.doExecuteInConnection(consDbConfig
          .getConsumerDs(), consDbConfig
          .getConsumerService(), consDbConfig
          .getConsumerDb(), connection -> {
            handleExceptionRestoreData(connection, dataInitContext);
            handleExceptionRestoreTable(connection, dataInitContext);
            return null;
          });
    } finally {
      this.handleExceptionCustomize(dataInitContext, ex);
    }

  }
  
  protected final String internalBackupThenExchangeTable(Connection connection, String productionTableName, String tempTableName, String backupTableSuffix) {
    return StringUtils.isEmpty(productionTableName) ? "" : (String)SqlUtils.executeWithInOnceStatement(connection, (stmt) -> {
      String backupTableName = "";
      String renameTempTableSql;
      if (SqlUtils.checkTableExistsWithInOnceStatement(connection, productionTableName)) {
        backupTableName = buildAdditionalTableName(productionTableName, backupTableSuffix);
        renameTempTableSql = SqlBuilder.buildRenameTable(productionTableName, backupTableName);
        stmt.execute(renameTempTableSql);
      }

      renameTempTableSql = SqlBuilder.buildRenameTable(tempTableName, productionTableName);
      stmt.execute(renameTempTableSql);
      return backupTableName;
    });
  }
  
  public static String buildAdditionalTableName(String masterTableName, String customSuffix) {
    Assert.notNull(masterTableName);
    Date timestamp = new Date();
    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
    String suffix = (String)Optional.<String>ofNullable(customSuffix).orElse("") + simpleDateFormat.format(timestamp);
    int maxAvailableTableLength = 30 - suffix.length();
    int masterTableNameEndIndex = Math.min(maxAvailableTableLength, masterTableName.length());
    String subMasterTableName = masterTableName.substring(0, masterTableNameEndIndex);
    return subMasterTableName + suffix;
  }
  
  protected abstract void preInitRunningStatus(DataInitContext paramDataInitContext);
  
  protected abstract void preInitTable(DataInitContext paramDataInitContext);
  
  protected abstract void preInitSuspendConsumerGroup(DataInitContext paramDataInitContext);
  
  protected abstract List<DataTransferContext> doInitializeBuildDataTransferContext(DataInitContext paramDataInitContext);
  
  protected abstract void postInitRefreshTargetTableCustomize(Connection paramConnection, DataInitContext paramDataInitContext) throws SQLException;
  
  protected abstract void postInitRefreshStatus(DataInitContext paramDataInitContext);
  
  protected abstract void handleExceptionRestoreData(Connection paramConnection, DataInitContext paramDataInitContext);
  
  protected abstract void handleExceptionRestoreTable(Connection paramConnection, DataInitContext paramDataInitContext);
  
  protected abstract void handleExceptionCustomize(DataInitContext paramDataInitContext, Exception paramException);
  
  public static class DataInitContext {
    private ProducerConfig producerConfig;
    
    private ConsDbConfig consDbConfig;
    
    private ConsTenantConfig consTenantConfig;
    
    private Date dataInitTimestamp;
    
    private AtomicBoolean stopInit;
    
    public DataInitContext(ProducerConfig producerConfig, ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
      this.stopInit = new AtomicBoolean(false);
      this.producerConfig = producerConfig;
      this.consDbConfig = consDbConfig;
      this.consTenantConfig = consTenantConfig;
    }
    
    public ProducerConfig getProducerConfig() {
      return this.producerConfig;
    }
    
    public DataInitContext setProducerConfig(ProducerConfig producerConfig) {
      this.producerConfig = producerConfig;
      return this;
    }
    
    public ConsDbConfig getConsDbConfig() {
      return this.consDbConfig;
    }
    
    public DataInitContext setConsDbConfig(ConsDbConfig consDbConfig) {
      this.consDbConfig = consDbConfig;
      return this;
    }
    
    public ConsTenantConfig getConsTenantConfig() {
      return this.consTenantConfig;
    }
    
    public DataInitContext setConsTenantConfig(ConsTenantConfig consTenantConfig) {
      this.consTenantConfig = consTenantConfig;
      return this;
    }
    
    public Date getDataInitTimestamp() {
      return this.dataInitTimestamp;
    }
    
    public DataInitContext setDataInitTimestamp(Date dataInitTimestamp) {
      this.dataInitTimestamp = dataInitTimestamp;
      return this;
    }
    
    public void stop() {
      this.stopInit.set(true);
    }
    
    public boolean canInit() {
      return !this.stopInit.get();
    }
    
    public String toString() {
      return (new ToStringBuilder(this))
        .append("producerConfig", this.producerConfig)
        .append("consDbConfig", this.consDbConfig)
        .append("consTenantConfig", this.consTenantConfig)
        .append("dataInitTimestamp", this.dataInitTimestamp)
        .append("stopInit", this.stopInit)
        .toString();
    }
  }
  
  public static class DataTransferContext {
    private DataInitContext dataInitContext;
    
    TableInitContext sourceTableInitContext;
    
    TableInitContext targetTableInitContext;
    
    public DataTransferContext(DataInitContext dataInitContext, TableInitContext sourceTableInitContext, TableInitContext targetTableInitContext) {
      this.dataInitContext = dataInitContext;
      this.sourceTableInitContext = sourceTableInitContext;
      this.targetTableInitContext = targetTableInitContext;
    }
    
    public void validate() {
      Assert.notNull(this.dataInitContext, "data init context cannot be null.");
      Assert.notNull(this.sourceTableInitContext, "source table init context cannot be null.");
      Assert.notNull(this.sourceTableInitContext.getInitFetchDataSql(), "source table init fetch data sql cannot be null.");
      Assert.notNull(this.sourceTableInitContext.getInitFetchDataSqlParameterValue(), "source table init fetch data sql parameter value cannot be null.");
      Assert.notNull(this.targetTableInitContext, "target table init context cannot be null.");
      Assert.notNull(this.targetTableInitContext.getInitQualifiedTableName(), "target table init qualified table name cannot be null.");
    }
    
    public DataInitContext getDataInitContext() {
      return this.dataInitContext;
    }
    
    public DataTransferContext setDataInitContext(DataInitContext dataInitContext) {
      this.dataInitContext = dataInitContext;
      return this;
    }
    
    public TableInitContext getSourceTableInitContext() {
      return this.sourceTableInitContext;
    }
    
    public DataTransferContext setSourceTableInitContext(TableInitContext sourceTableInitContext) {
      this.sourceTableInitContext = sourceTableInitContext;
      return this;
    }
    
    public TableInitContext getTargetTableInitContext() {
      return this.targetTableInitContext;
    }
    
    public DataTransferContext setTargetTableInitContext(TableInitContext targetTableInitContext) {
      this.targetTableInitContext = targetTableInitContext;
      return this;
    }
    
    public String toString() {
      return (new ToStringBuilder(this))
        .append("dataInitContext", this.dataInitContext)
        .append("sourceTableInitContext", this.sourceTableInitContext)
        .append("targetTableInitContext", this.targetTableInitContext)
        .toString();
    }
  }
  
  public static class TableInitContext {
    private String initDataSourceCode;
    
    private String initServiceName;
    
    private String initDbcatalog;
    
    private String initTableName;
    
    private String initQualifiedTableName;
    
    private String initTableDdl;
    
    private String initTableDdlPlaceholder;
    
    private String initFetchDataSql;
    
    private Object initFetchDataSqlParameterValue;
    
    public TableInitContext(String initDataSourceCode, String initServiceName, String initDbcatalog, String initTableName, String initTableDdl, String initTableDdlPlaceholder) {
      this.initDataSourceCode = initDataSourceCode;
      this.initServiceName = initServiceName;
      this.initDbcatalog = initDbcatalog;
      this.initTableName = initTableName;
      this.initTableDdl = initTableDdl;
      this.initTableDdlPlaceholder = initTableDdlPlaceholder;
    }
    
    public void validate() {
      Assert.notNull(this.initDataSourceCode, "init data source id cannot be null.");
      Assert.notNull(this.initServiceName, "init service name cannot be null.");
      Assert.notNull(this.initDbcatalog, "init db catalog cannot be null");
      Assert.notNull(this.initTableDdl, "init table ddl cannot be null.");
      Assert.notNull(this.initTableDdlPlaceholder, "init table ddl table name placeholder cannot be null.");
      Assert.notNull(this.initTableName, "init table name cannot be null.");
    }
    
    public void doInit(TransferDataSourceTemplate transferDataSourceTemplate, CheckTableHandler checkTableHandler) {
      AbstractDataInitService.LOGGER.info("begin init table {}.", toString());
      validate();
      transferDataSourceTemplate.doExecuteInConnection(
          getInitDataSourceCode(), 
          getInitServiceName(), 
          getInitDbcatalog(), connection -> {
            doInitWithInConnection(connection, checkTableHandler);
            return null;
          });
    }
    
    public void doInitWithInConnection(Connection connection, CheckTableHandler checkTableHandler) {
      AbstractDataInitService.LOGGER.info("begin init table in connection {}.", toString());
      validate();
      try {
        if (SqlUtils.checkTableExistsWithInOnceStatement(connection, getInitTableName())) {
          checkTableHandler.handleWhenCheckTableExist(connection, this);
        } else {
          checkTableHandler.handleWhenCheckTableNoneExist(connection, this);
        } 
      } catch (SQLException ex) {
        AbstractDataInitService.LOGGER.error("init table with error: ");
        throw new CommonException(ex);
      } 
    }
    
    public static TableInitContext buildProducerDBTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      if (!StringUtils.isEmpty(producerConfig.getTableName())) {
        return new TableInitContext(producerConfig
            .getInitDsCode(), producerConfig
            .getServiceCode(), producerConfig
            .getInitDbCode(), producerConfig
            .getTableName(), producerConfig
            .getMainTableDDL(), producerConfig
            .getTableName());
      }
      return null;
    }
    
    public static TableInitContext buildProducerDBTlTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      if (!StringUtils.isEmpty(producerConfig.getTlTableName())) {
        return new TableInitContext(producerConfig
            .getInitDsCode(), producerConfig
            .getServiceCode(), producerConfig
            .getInitDbCode(), producerConfig
            .getTlTableName(), producerConfig
            .getTlTableDDL(), producerConfig
            .getTlTableName());
      }
      return null;
    }
    
    public static TableInitContext buildConsumerDBTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
      if (!StringUtils.isEmpty(producerConfig.getTableName())) {
        return new TableInitContext(consDbConfig
            .getConsumerDs(), consDbConfig
            .getConsumerService(), consDbConfig
            .getConsumerDb(), consDbConfig
            .getConsumerTable(), producerConfig
            .getMainTableDDL(), producerConfig
            .getTableName());
      }
      return null;
    }
    
    public static TableInitContext buildConsumerDBTlTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
      if (!StringUtils.isEmpty(producerConfig.getTlTableName())) {
        return new TableInitContext(consDbConfig
            .getConsumerDs(), consDbConfig
            .getConsumerService(), consDbConfig
            .getConsumerDb(), consDbConfig
            .getConsumerTlTable(), producerConfig
            .getTlTableDDL(), producerConfig
            .getTlTableName());
      }
      return null;
    }
    
    public static TableInitContext buildConsumerDBTempTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
      if (!StringUtils.isEmpty(producerConfig.getTableName())) {
        if (StringUtils.isEmpty(producerConfig.getInitTempTable())) {
          producerConfig.setInitTempTable(AbstractDataInitService.buildAdditionalTableName(producerConfig.getTableName(), null));
        }
        return new TableInitContext(consDbConfig
            .getConsumerDs(), consDbConfig
            .getConsumerService(), consDbConfig
            .getConsumerDb(), producerConfig
            .getInitTempTable(), producerConfig
            .getMainTableDDL(), producerConfig
            .getTableName());
      } 
      return null;
    }
    
    public static TableInitContext buildConsumerDBTempTlTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
      if (!StringUtils.isEmpty(producerConfig.getTlTableName())) {
        if (StringUtils.isEmpty(producerConfig.getInitTempTlTable())) {
          producerConfig.setInitTempTlTable(AbstractDataInitService.buildAdditionalTableName(producerConfig.getTlTableName(), "_tl"));
        }
        return new TableInitContext(consDbConfig
            .getConsumerDs(), consDbConfig
            .getConsumerService(), consDbConfig
            .getConsumerDb(), producerConfig
            .getInitTempTlTable(), producerConfig
            .getTlTableDDL(), producerConfig
            .getTlTableName());
      } 
      return null;
    }
    
    public static TableInitContext buildConsumerDBBackupTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
      if (StringUtils.isEmpty(producerConfig.getTableName())) {
        if (StringUtils.isEmpty(producerConfig.getBackupTableName())) {
          producerConfig.setBackupTableName(AbstractDataInitService.buildAdditionalTableName(producerConfig.getTableName(), "_bak"));
        }
        return new TableInitContext(consDbConfig
            .getConsumerDs(), consDbConfig
            .getConsumerService(), consDbConfig
            .getConsumerDb(), producerConfig
            .getBackupTableName(), producerConfig
            .getMainTableDDL(), producerConfig
            .getTableName());
      } 
      return null;
    }
    
    public static TableInitContext buildConsumerDBBackupTlTable(DataInitContext dataInitContext) {
      ProducerConfig producerConfig = dataInitContext.getProducerConfig();
      ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
      if (!StringUtils.isEmpty(producerConfig.getTlTableName())) {
        if (StringUtils.isEmpty(producerConfig.getBackupTlTableName())) {
          producerConfig.setBackupTlTableName(AbstractDataInitService.buildAdditionalTableName(producerConfig.getTlTableName(), "_bak"));
        }
        return new TableInitContext(consDbConfig
            .getConsumerDs(), consDbConfig
            .getConsumerService(), consDbConfig
            .getConsumerDb(), producerConfig
            .getBackupTlTableName(), producerConfig
            .getTlTableDDL(), producerConfig
            .getTlTableName());
      } 
      return null;
    }
    
    public TableInitContext buildQualifiedTableName() {
      this.initQualifiedTableName = SqlUtils.buildQualifiedTableName(
          getInitDataSourceCode(), 
          getInitServiceName(), 
          getInitDbcatalog(), 
          getInitTableName());
      return this;
    }
    
    public TableInitContext buildQualifiedTableName(Connection connection) {
      this.initQualifiedTableName = SqlUtils.buildQualifiedTableNameInConnection(connection, 
          
          getInitDbcatalog(), 
          getInitTableName());
      return this;
    }
    
    public TableInitContext buildInitFetchDataSql(List<String> whereFieldList, String appendingCustomSqlRawSlice, Object initFetchDataSqlParameterValue) {
      this.initFetchDataSql = SqlBuilder.buildSelectAllColumn(whereFieldList, 
          
          getInitQualifiedTableName(), appendingCustomSqlRawSlice);
      this.initFetchDataSqlParameterValue = initFetchDataSqlParameterValue;
      return this;
    }
    
    public String getInitServiceName() {
      return this.initServiceName;
    }
    
    public TableInitContext setInitServiceName(String initServiceName) {
      this.initServiceName = initServiceName;
      return this;
    }
    
    public String getInitDbcatalog() {
      return this.initDbcatalog;
    }
    
    public TableInitContext setInitDbcatalog(String initDbcatalog) {
      this.initDbcatalog = initDbcatalog;
      return this;
    }
    
    public String getInitTableName() {
      return this.initTableName;
    }
    
    public TableInitContext setInitTableName(String initTableName) {
      this.initTableName = initTableName;
      return this;
    }
    
    public String getInitQualifiedTableName() {
      return this.initQualifiedTableName;
    }
    
    public String getInitTableDdl() {
      return this.initTableDdl;
    }
    
    public TableInitContext setInitTableDdl(String initTableDdl) {
      this.initTableDdl = initTableDdl;
      return this;
    }
    
    public String getInitTableDdlPlaceholder() {
      return this.initTableDdlPlaceholder;
    }
    
    public TableInitContext setInitTableDdlPlaceholder(String initTableDdlPlaceholder) {
      this.initTableDdlPlaceholder = initTableDdlPlaceholder;
      return this;
    }
    
    public String getInitFetchDataSql() {
      return this.initFetchDataSql;
    }
    
    public Object getInitFetchDataSqlParameterValue() {
      return this.initFetchDataSqlParameterValue;
    }
    
    public String toString() {
      return (new ToStringBuilder(this))
        .append("initDataSourceCode", this.initDataSourceCode)
        .append("initServiceName", this.initServiceName)
        .append("initDbcatalog", this.initDbcatalog)
        .append("initTableName", this.initTableName)
        .append("initQualifiedTableName", this.initQualifiedTableName)
        .append("initTableDdl", this.initTableDdl)
        .append("initTableDdlPlaceholder", this.initTableDdlPlaceholder)
        .append("initFetchDataSql", this.initFetchDataSql)
        .append("initFetchDataSqlParameterValue", this.initFetchDataSqlParameterValue)
        .toString();
    }
    
    public String getInitDataSourceCode() {
      return this.initDataSourceCode;
    }
    
    public void setInitDataSourceCode(String initDataSourceCode) {
      this.initDataSourceCode = initDataSourceCode;
    }
  }
  
  @FunctionalInterface
  public static interface CheckTableHandler {
    void handleWhenCheckTableExist(Connection param1Connection, TableInitContext param1TableInitContext) throws SQLException;
    
    default void handleWhenCheckTableNoneExist(Connection connection, TableInitContext tableInitContext) throws SQLException {
      SqlUtils.executeWithInOnceStatement(connection, statement -> {
            String ddl = tableInitContext.getInitTableDdl().replace(tableInitContext.getInitTableDdlPlaceholder(), tableInitContext.getInitTableName());
            AbstractDataInitService.LOGGER.info("handle when check table {} non exist with ddl {}.", tableInitContext.getInitTableName(), ddl);
            statement.execute(ddl);
            return null;
          });
    }
  }
}
